package david.java.flink_sql.udf.offline;

import org.apache.hadoop.hive.ql.exec.UDF;

import java.util.*;

/**
 * @Description: 本udf函数是用来计算表血缘关系的, 该方法遍历方式为 DFS, 不适合大数量, 递归层数较深的场景. 如果有大数据量或递归层数较深的场景,
 * 请不要使用本方法, 请使用BFS遍历方式实现
 *
 * @Author: zhaodawei
 * @Date: Create in 3:37 下午 2022/3/8
 */
public class GetSerial extends UDF {

    public List<String> evaluate(List<String> list) {
        Map<String, HashSet<String>> res = new HashMap<String, HashSet<String>>();
        Map<String, HashSet<String>> tmp = new HashMap<String, HashSet<String>>();
        for (String value : list) {
            String[] a = value.split("-");
            HashSet<String> s = new HashSet<String>();
            if (tmp.containsKey(a[0])) {
                s = tmp.get(a[0]);
                s.add(a[1]);
                tmp.put(a[0], s);
            } else {
                s.add(a[1]);
                tmp.put(a[0], s);
            }
        }

        for (Map.Entry<String, HashSet<String>> map : tmp.entrySet()) {
            String key = map.getKey();
            HashSet<String> result = new HashSet<String>();
            recursive(tmp, key, result, new HashSet<String>());
            result.remove(key);
            res.put(key, result);
        }

        return getList(res);
    }

    public void recursive(Map<String, HashSet<String>> data, String value, HashSet<String> result, HashSet<String> storedKeys) {
        HashSet<String> valueSet = data.get(value);
        storedKeys.add(value);
        if (valueSet == null) {
            result.add(value);
            return;
        }
        // 排除掉环
        for (String str : valueSet) {
            if (!storedKeys.contains(str)) {
                recursive(data, str, result, storedKeys);
            }
        }
        result.add(value);
    }

    public List<String> getList(Map<String, HashSet<String>> input) {
        List<String> res = new ArrayList<String>();
        for (Map.Entry<String, HashSet<String>> str : input.entrySet()) {
            String key = str.getKey();
            HashSet<String> values = str.getValue();
            for (String value : values) {
                res.add(key.concat("-").concat(value));
            }
        }
        return res;
    }

    public static void main(String[] args) {
        GetSerial gts = new GetSerial();
        List<String> list1 = new ArrayList<String>();
        list1.add("a-b");
        list1.add("b-a");
        list1.add("b-c");
        list1.add("c-b");
        list1.add("c-d");
        list1.add("d-c");
        System.out.println(gts.evaluate(list1));
        System.out.println();

        List<String> list2 = new ArrayList<String>();
        list2.add("a-b");
        list2.add("b-c");
        list2.add("c-a");
        System.out.println(gts.evaluate(list2));
        System.out.println();


        List<String> list3 = new ArrayList<String>();
        list3.add("a-b");
        list3.add("b-c");
        list3.add("c-d");
        System.out.println(gts.evaluate(list3));
        System.out.println();

    }

}





